package kafka;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import job.KafkaProperties;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import utils.SystemParam.Topic;

public final class ConsumerFactory {
	
	/**
	 *  创建到zookeeper的连接
	 * @return ConsumerConnector
	 */
	public static ConsumerConnector createConsumerConnector() {
		Properties props = new Properties();
		props.put("zookeeper.connect", KafkaProperties.zkConnect);// 要连接的zookeeper地址
		props.put("group.id", KafkaProperties.groupId);// 消费者所在的组
//		props.put("conusmer.id", KafkaProperties.consumerId);// 此消费者的唯一标识用于在zookeeper中存储此消费者的偏移量
		//props.put("conusmer.id", "saleConsumer");// 此消费者的唯一标识用于在zookeeper中存储此消费者的偏移量
		props.put("zookeeper.session.timeout.ms", KafkaProperties.zSessionTimeout);// zookeeper连接超时时间
		props.put("auto.offset.reset", "smallest");
		props.put("auto.commit.enable", "false");
		ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
		return consumer;
	}
	/**
	 * 取得某一个消息的流
	 * @param topic
	 * @param consumer
	 * @return KafkaStream<String, String> 
	 */
	public static KafkaStream<String, String> createKafkaStream(Topic topic,ConsumerConnector consumer){
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic.name(), new Integer(1));// topic:要取的主题，Integer:创建的数据流个数
		System.out.println("before create message stream");
		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap, new StringDecoder(null), new StringDecoder(null));// 获取创建的流
		System.out.println("after create message stream");
		List<KafkaStream<String, String>> list = consumerMap.get(topic.name());// 取得某一个消息的流
		System.out.println("after get topic");
		KafkaStream<String, String> kafkaStream = list.get(0);
		return kafkaStream;
	}
	/**
	 * 取得某一个消息的流
	 * @param topic
	 * @param consumer
	 * @return KafkaStream<String, String> 
	 */
	public static KafkaStream<String, String> createKafkaStream(String topic,ConsumerConnector consumer){
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, new Integer(1));// topic:要取的主题，Integer:创建的数据流个数
		System.out.println("before create message stream");
		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap, new StringDecoder(null), new StringDecoder(null));// 获取创建的流
		System.out.println("after create message stream");
		List<KafkaStream<String, String>> list = consumerMap.get(topic);// 取得某一个消息的流
		System.out.println("after get topic");
		KafkaStream<String, String> kafkaStream = list.get(0);
		return kafkaStream;
	}
}
